package com.ruoyi.model.config;

import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.toolkit.sql.SqlUtils;
import com.ruoyi.common.utils.DateUtils;
import com.ruoyi.model.domain.ModelTaskProcess;
import com.ruoyi.model.domain.vo.ModelTaskProcessVo;
import com.ruoyi.model.domain.vo.ModelTaskVo;
import com.ruoyi.model.service.IModelTaskProcessService;
import com.ruoyi.model.service.IModelTaskService;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.mqtt.support.MqttHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

/**
 * mqtt接收配置
 * @author liuxiaojie
 * @date 2022/3/30 13:52
 */
@Configuration
@IntegrationComponentScan
public class MqttReceiveConfig {
    @Value("${spring.mqtt.username}")
    private String username;

    @Value("${spring.mqtt.password}")
    private String password;

    @Value("${spring.mqtt.url}")
    private String hostUrl;

    @Value("${spring.mqtt.clientId}")
    private String clientId;

    @Value("${spring.mqtt.default.topic}")
    private String defaultTopic;

    @Value("${spring.mqtt.default.completionTimeout}")
    private long completionTimeout;

    @Value("${spring.mqtt.service}")
    private String mqttservice;


    @Autowired
    private IModelTaskProcessService modelTaskProcessService;

    @Autowired
    private IModelTaskService modelTaskService;


    @Bean
    public MqttConnectOptions getMqttConnectOptions(){
        MqttConnectOptions mqttConnectOptions=new MqttConnectOptions();
        mqttConnectOptions.setUserName(username);
        mqttConnectOptions.setPassword(password.toCharArray());
        mqttConnectOptions.setServerURIs(new String[]{hostUrl});
        mqttConnectOptions.setKeepAliveInterval(20);
        return mqttConnectOptions;
    }
    @Bean("mls")
    public MqttPahoClientFactory mqttClientFactory() {
        DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(getMqttConnectOptions());
        return factory;
    }

    /**
     * 配置client,监听的topic
     *
     * @param messageChannel
     * @return
     */
    @Bean("mls-inbound")
    public MessageProducer inbound(@Qualifier("mls-mqttInputChannel") MessageChannel messageChannel) {
        String[] topics=defaultTopic.trim().split(",");
        //这里的defaultTopic是发布者的主题
        MqttPahoMessageDrivenChannelAdapter adapter =
                new MqttPahoMessageDrivenChannelAdapter(clientId + System.currentTimeMillis(), mqttClientFactory(),topics);
        adapter.setCompletionTimeout(completionTimeout);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        adapter.setOutputChannel(messageChannel);
        return adapter;
    }

    /**
     *
     ServiceActivator注解表明当前方法用于处理MQTT消息，inputChannel参数指定了用于接收消息信息的channel。
     * @return
     */
    @Bean("mls-handler")
    @ServiceActivator(inputChannel = "mls-mqttInputChannel")
    public MessageHandler handler() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                String topic=message.getHeaders().get(MqttHeaders.RECEIVED_TOPIC).toString();
                System.out.println(topic);
                String msg=message.getPayload().toString();
                String taskExecTopic="/model-task/exec";
                String taskResultTopic="/model-task/result";
                String stroeTopic="mls/agribusiness/";
                String specialistTopic="mls/specialist/";

                if(StrUtil.startWith(topic,mqttservice+taskExecTopic)){
                    HashMap data= JSONUtil.toBean(msg,HashMap.class);
                    System.out.println(data);

                }else if(StrUtil.startWith(topic,mqttservice+taskResultTopic)){
                    ModelTaskProcessVo process=JSONUtil.toBean(msg,ModelTaskProcessVo.class);
                    ModelTaskVo taskVo=modelTaskService.selectModelTaskVoById(process.getTaskId());
                    int totals=taskVo.getProcessList().size();
                    int finish=0;
                    int error=0;
                    for(ModelTaskProcessVo taskProcessVo:taskVo.getProcessList()){
                       if(taskProcessVo.getId().equals(process.getId())){
                           taskProcessVo.setProcessStatus(process.getProcessStatus());
                           modelTaskProcessService.updateModelTaskProcess(taskProcessVo);
                       }
                       int status=Integer.parseInt(taskProcessVo.getProcessStatus());
                       if(status>2){
                           error++;
                           finish++;
                       }else if(status==2){
                           finish++;
                       }
                    }
                    if(totals==finish){
                        taskVo.setEndTime(DateUtils.dateTimeNow("yyyy-MM-dd HH:mm:ss"));
                        if(error>0){
                            taskVo.setTaskStatus("3");
                        }else{
                            taskVo.setTaskStatus("2");
                        }
                        modelTaskService.updateModelTask(taskVo);
                    }

                }else if(StrUtil.startWith(topic,stroeTopic)){

                }else if(StrUtil.startWith(topic,specialistTopic)){

                }
            }
        };
    }


    @Bean("mls-mqttInputChannel")
    public MessageChannel mqttInputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageChannel mqttOutboundChannel() {
        return new DirectChannel();
    }

    /**
     * ServiceActivator注解表明当前方法用于处理MQTT消息，inputChannel参数指定了用于接收消息信息的channel。
     * @return
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttOutboundChannel")
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler =  new MqttPahoMessageHandler(clientId+System.currentTimeMillis(), mqttClientFactory());
        //如果设置成true，发送消息时将不会阻塞。
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic(defaultTopic);
        return messageHandler;
    }
}
